﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Messaging;
using System.Threading;
using System.Runtime.Remoting.Messaging;

namespace Niknak.Services.ServiceCore.Msmq
{
    public abstract class Processor<ConfigurationSettings, ExceptionHandler> 
        where ConfigurationSettings : IMsmqConfig, new()
        where ExceptionHandler : IExceptionHandler, new()
    {
        protected static int _ThreadCount = 0;
        protected bool _SetMaxThreadsResult;
        protected bool _ShuttingDown;
        protected ProcessMessage _ProcessMessage;
        public static object OpLocker = new object();

        public static ConfigurationSettings Settings { get { return new ConfigurationSettings(); } }
        public static ExceptionHandler CustomExceptionHandler { get { return new ExceptionHandler(); } }

        private static List<ThreadedOperation> _OpStack = null;
        public static List<ThreadedOperation> OpStack
        {
            get
            {
                if (_OpStack == null)
                    _OpStack = new List<ThreadedOperation>();
                return _OpStack;
            }
        }

        public static bool PushOp(ThreadedOperation op)
        {
            lock (OpLocker)
            {
                if (OpStack.Contains(op))
                    return false;

                OpStack.Add(op);
                Console.WriteLine("Pushed operation to stack. Count is now " + OpStack.Count.ToString());
                return true;
            }
        }

        public static bool PopOp(ThreadedOperation op)
        {
            lock (OpLocker)
            {
                if (!OpStack.Contains(op))
                    return false;

                OpStack.Remove(op);
                Console.WriteLine("Popped operation from stack. Count is now " + OpStack.Count.ToString());
                return true;
            }
        }

        public virtual void Initialise()
        {
            if (_ProcessMessage == null)
                throw new Exception("Message processor delegate is undefined");

            Helper<ConfigurationSettings, ExceptionHandler>.QueueReader.ReceiveCompleted += new ReceiveCompletedEventHandler(MessageReceiveComplete);

            _SetMaxThreadsResult = ThreadPool.SetMaxThreads(
                   Settings.MaxMessageProcessorThreads,
                   Settings.MaxMessageProcessorThreads
               );

            if (!_SetMaxThreadsResult)
                throw new Exception("Unable to set threadpool limits");

            StartListening();
        }

        private void MessageReceiveComplete(object sender, ReceiveCompletedEventArgs e)
        {
            if (_ShuttingDown)
                return;
            
            //queue that has received a message
            MessageQueue sourceQueue = (MessageQueue)sender;
            try
            {
                while (_ThreadCount >= Settings.MaxMessageProcessorThreads)
                    Thread.Sleep(50);

                //complete the message receive
                Message queuedMessage = sourceQueue.EndReceive(e.AsyncResult);

                //throttling
                if (Settings.ThrottleDurationSeconds > 0 && Settings.MaxMessagesHandledPerThrottleDuration > 0)
                {
                    if (Throttle<ConfigurationSettings>.Instance.IsThrottled((Enums.Msmq.MessageType)Enum.Parse(typeof(Enums.Msmq.MessageType), queuedMessage.Label, false)))
                    {
                        //put the message back on the queue with an elevated priority - todo, make the reading of priority work properly
                        Helper<ConfigurationSettings, ExceptionHandler>.Send((Enums.Msmq.MessageType)Enum.Parse(typeof(Enums.Msmq.MessageType), queuedMessage.Label, false), MessagePriority.AboveNormal, queuedMessage.Body);

                        int throttleRetry = Settings.ThrottleRetrySeconds;
                        if (throttleRetry <= 0)
                            throttleRetry = 1;

                        Thread.Sleep(TimeSpan.FromSeconds(throttleRetry));
                        StartListening();
                        Console.WriteLine("Throttled");
                        return;
                    }
                    Throttle<ConfigurationSettings>.Instance.Add(queuedMessage);
                }

                IAsyncResult asyncResult = _ProcessMessage.BeginInvoke(
                    queuedMessage,
                    ProcessMessageCallback,
                    null
                );
                _ThreadCount++;

                StartListening();
                Console.WriteLine("Spun thread for message worker");
            }
            catch (MessageQueueException ex)
            {
                switch ((int)ex.MessageQueueErrorCode)
                {
                    case -1073741536: //the error that doesn't exist - but usually just means the queue couldn't be reached
                        Thread.Sleep(5000);
                        StartListening();
                        break;
                    default:
                        CustomExceptionHandler.ExceptionLogEntry(ex);
                        break;
                }
            }
            catch (Exception ex)
            {
                CustomExceptionHandler.ExceptionLogEntry(ex);
            }
        }

        private void ProcessMessageCallback(IAsyncResult result)
        {
            Console.WriteLine("Handling callback");

            ProcessMessage worker = (ProcessMessage)((AsyncResult)result).AsyncDelegate;
            OperationResult returnValue = worker.EndInvoke(result);
            _ThreadCount--;

            switch (returnValue.Result)
            {
                case InternalEnums.Msmq.ResultStatus.Undetermined:
                case InternalEnums.Msmq.ResultStatus.Fail:
                    //opportunity for additional action/logging at the service level
                    break;
            }
        }

        protected void StartListening()
        {
            Helper<ConfigurationSettings, ExceptionHandler>.QueueReader.Refresh();
            Helper<ConfigurationSettings, ExceptionHandler>.QueueReader.BeginReceive();
        }

        public void Shutdown()
        {
            _ShuttingDown = true;
            Helper<ConfigurationSettings, ExceptionHandler>.QueueReader.Close();
        }

        protected delegate OperationResult ProcessMessage(Message queuedMessage);
    }
}
